<?php
	namespace ampleharvest\task\index_data;

	use\ampleharvest\code\cRestConnector;
	use\ampleharvest\code\cConfig;
	use\ampleharvest\code\cRestServices;
	use\ampleharvest\code;

	require_once dirname(dirname(dirname(dirname(__FILE__)))) . "/config/cConfig.php";
	require_once dirname(dirname(dirname(dirname(__FILE__)))) . "/config/cRestServices.php";
	require_once dirname(dirname(dirname(dirname(__FILE__)))) . "/classes/cRestConnector.php";
	require_once dirname(dirname(dirname(dirname(__FILE__)))) . "/abstract_classes/abs_Task.php";
	require_once dirname(dirname(dirname(dirname(__FILE__)))) . "/classes/cThread.php";
	require_once dirname(dirname(dirname(dirname(__FILE__)))) . '/classes/cStomp.php';
	require_once dirname(dirname(dirname(dirname(__FILE__)))) . '/classes/search/solr/cSolr.php';

	$arr_Params = array("id::",
		"optimize::");

	$arr_Options = getopt("c::", $arr_Params);
	/**
	 * @brief This Task Handles pulling ids of objects off a Queue and pulling out important information then putting the ids in a queue to be Indexed
	 *
	 * - Revision History
	 *  - 0.1
	 * 	 - <b>Date</b>: 02 SEP 2011
	 * 	 - <b>Author</b>: Randy Blancett
	 * 	  - Initial Version
																												
	 * @author Randy Blancett
	 * @version 0.1
	 * @ingroup code
																																												
	 */
	class index_Data_Queue extends \abs_Task
	{
		const C_STR_NAME = "index_Data_Queue";
		const C_INT_THREADS = 1;
		const C_INT_COMMIT_NUMBER = 200;
		const C_INT_OPTIMIZE_NUMBER = 1000;
		const C_INT_RETRY_LIMIT = 5;

		protected $m_arr_Threads = null;
		protected $arr_Options = null;

		protected $m_int_Commit_Number = 0;
		protected $m_int_Optimize_Number = 0;
		
		protected $m_bool_Optimize = true;

		protected static $m_obj_SOLR = null;
		protected $m_obj_STOMP = null;

		protected $m_obj_RestCon = null;

		public function __construct($str_LockPath, $arr_Options = array())
		{
			$this->set_LockPath($str_LockPath);
			$this->set_Name(self::C_STR_NAME);
			$this->m_arr_Threads = array();
			$this->arr_Options = $arr_Options;
			
			if(isset($arr_Options['optimize']))
			{
				if(!$arr_Options['optimize'])
				{
					$this->m_bool_Optimize = false;
					print("No Optimze\n");
				}
			}
			
			parent::__construct();
			$this->set_MaxThreads(self::C_INT_THREADS);
		}

		protected static function get_SOLR()
		{
			if (!self::$m_obj_SOLR)
			{
				self::$m_obj_SOLR = new \cSolr(null, null, null);
			   $this->output_Error("Using default solr instance -- index_Data_Queue");
			}
			return self::$m_obj_SOLR;
		}

		protected function get_STOMP()
		{
			if (!$this->m_obj_STOMP)
			{
				$this->m_obj_STOMP = new \cStomp();
				$this->m_obj_STOMP->subscribe(code\cConfig::C_STR_SOLR_QUEUE);
			}

			return $this->m_obj_STOMP;
		}

		protected function get_RestCon()
		{
			if (!$this->m_obj_RestCon)
			{
				$this->m_obj_RestCon = new cRestConnector("");
				$this->m_obj_RestCon->set_Login(cConfig::C_STR_AMPLE_SILO_USER_NAME, cConfig::C_STR_AMPLE_SILO_PASSWORD);
			}

			return $this->m_obj_RestCon;
		}

		public function run($bool_Force = false)
		{
			//print("Test".__LINE__."\n");
			$this->set_Force($bool_Force);
			if (!parent::run())
			{
				return false;
			}

			$this->startThreads();

			$this->wait4Threads();
			
			$this->solr_Commit();

			$this->solr_Optimize();

			return true;
		}

		private function solr_Commit()
		{
			$obj_SOLR = self::get_SOLR();

			try
			{
				$obj_SOLR->commit();
				print("Solr Commited" . PHP_EOL);
			}
			catch (Exception $e)
			{
				$this->log_Error("encountered the following problem when trying to comit the solr updates: " . $e);
				return;
			}

			$this->m_int_Commit_Number = 0;
		}

		private function solr_Optimize()
		{
			if(!$this->m_bool_Optimize)
			{
				return;
			}
			$obj_SOLR = self::get_SOLR();

			try
			{
				$obj_SOLR->optimize();
				print("Solr Optimized" . PHP_EOL);
			}
			catch (Exception $e)
			{
				$this->log_Error("encountered the following problem when trying to optimize solr: " . $e);
				return;
			}

			$this->m_int_Optimize_Number = 0;
		}

		private function startThreads()
		{
			while (sizeof($this->m_arr_Threads) <= self::C_INT_THREADS) // ($int_I = 0; $int_I < self::C_INT_THREADS; $int_I++)
			{
				$this->killDeadThreads();
				if (!$this->startThread())
				{
					$int_Count = 1;
					$bool_Kill = true;
					while ($int_Count < self::C_INT_RETRY_LIMIT)
					{
						if (!$this->startThread())
						{
							$int_Count++;
						}
						else
						{
							$bool_Kill = false;
							break;
						}
					}

					if ($bool_Kill)
					{
						break;
					}
				}
			}
		}

		private function startThread()
		{
			$this->m_int_Commit_Number ++;
			$this->m_int_Optimize_Number ++;

			$obj_STOMP = $this->get_STOMP();

			$obj_Msg = $obj_STOMP->getMessage();

			if (!$obj_Msg)
			{
				return false;
			}

			$str_ID = $obj_Msg->body;

			//print("Doc ID - " . $str_ID . "\n");

			$str_Command = __FILE__ . " -c --id=\"" . $str_ID . "\"";
			print($str_Command . "\n");
			$int_I = sizeof($this->m_arr_Threads);
			$this->m_arr_Threads[$int_I] = \cThread::Create($str_Command, code\cConfig::C_STR_PATH_PHP);

			$obj_STOMP->ackMessage($obj_Msg);

			return true;
		}

		private function killDeadThreads()
		{
			foreach ($this->m_arr_Threads as $str_Key => $obj_Thread)
			{
				if (!$obj_Thread->isActive())
				{
					$str_Return = $obj_Thread->listen();
					print($str_Return);
					$obj_Thread->close();
					unset($this->m_arr_Threads[$str_Key]);
					if ($this->m_int_Commit_Number >= self::C_INT_COMMIT_NUMBER)
					{
						$this->solr_Commit();
					}

					if ($this->m_int_Optimize_Number >= self::C_INT_OPTIMIZE_NUMBER)
					{
						$this->solr_Optimize();
					}
				}
			}
		}

		private function wait4Threads()
		{
			while (sizeof($this->m_arr_Threads) > 0)
			{ //No Items to process
				$this->killDeadThreads();

				$this->startThreads();
			}
		}

		public function runChunk()
		{
			$str_ID = $this->arr_Options["id"];
			$obj_AH_Doc = null;
			$obj_AH_Views = null;
			$obj_AH_Likes = null;
			$obj_AH_Tags = null;
			$arr_Data = array();

			$obj_RestCon = $this->get_RestCon();
			$obj_RestCon->set_Url(cRestServices::get_AHDataService($str_ID));

			$str_Contents = $obj_RestCon->send_Get();
			
			if(!$str_Contents)
			{
				print("Data server returned - ".$obj_RestCon->getCode()." - ".$obj_RestCon->getLastError().
				"for id $str_ID\n");
				return;
			}

			$obj_RestCon->clear_Params();

			try
			{
				$obj_AH_Doc = json_decode($str_Contents);
			}
			catch (Exception $e)
			{
				$this->set_Invalid($str_ID);
				$this->log_Error($str_ID . " the following problem unserializeing: " . $e);
				return;
			}
			//Get Views
			$obj_RestCon->set_Url(cRestServices::get_Views($str_ID));

			$str_Contents = $obj_RestCon->send_Get();

			$obj_RestCon->clear_Params();

			if ($str_Contents)
			{
				try
				{
					$obj_AH_Views = json_decode($str_Contents);
				}
				catch (Exception $e)
				{

				}
				
				if ($obj_AH_Views)
				{
					$obj_AH_Doc->m_obj_Views = $obj_AH_Views;
				}
			}

			$obj_SolrDoc = null;

			try
			{
				$obj_SolrDoc = \cSolr::load_AHDoc($obj_AH_Doc);
			}
			catch (Exception $e)
			{
				$this->set_Invalid($str_ID);
				$this->log_Error($str_ID . " the following problem: " . $e);
				return;
			}

			if ($obj_SolrDoc)
			{
				$obj_SOLR = self::get_SOLR();

				try
				{
					$obj_SOLR->addDocuments(array($obj_SolrDoc));
					//$obj_SOLR->commit();
				}
				catch (Exception $e)
				{
					$this->set_Invalid($str_ID);
					$this->log_Error($str_ID . " encountered the following problem when trying to add a solr document to the index: " . $e);
					return;
				}
			}
			else
			{
				$this->set_Invalid($str_ID);
				$this->log_Error($str_ID . " did not return a solr document for and unknown reason.");
			}
			$obj_Date = new \DateTime();

			$obj_RestCon->set_Param("date", $obj_Date->format("U"));
			$obj_RestCon->set_Param("id", $obj_AH_Doc->id);

			//print_r($arr_Data);

			$obj_RestCon->set_Url(cRestServices::get_AHIndex($str_ID));
			$obj_RestCon->send_Put();

			//cRestServices::send_Put(cRestServices::get_AHIndex($str_ID), $arr_Data);
		}

		private function set_Invalid($str_ID)
		{
			$obj_RestCon = $this->get_RestCon();
			$obj_RestCon->set_Url(cRestServices::get_AHIndex($str_ID));
			$obj_RestCon->clear_Params();

			$obj_RestCon->set_Param("error", 1);
			$obj_RestCon->set_Param("id", $str_ID);
			$obj_RestCon->set_Param("date", 1);

			//$arr_Data = array("error" => 1, "id" => $str_ID, "date" => 1);

			//print_r($arr_Data);

			print("Send invalid" . PHP_EOL);

			$obj_RestCon->send_Put();
			//cRestServices::send_Put(cRestServices::get_AHIndex($str_ID), $arr_Data);
		}

	}

	if (isset($arr_Options["c"]))
	{
		$obj_Task = new index_Data_Queue(null, $arr_Options);
		$obj_Task->runChunk();
	}
